#pragma once

#include "MessageHeader.h"
#include "MessageDispatcher.h"
#include "RedisPool.h"
#include <string>
#include <thread>
#include <atomic>
#include <vector>
#include <functional>

using TraderEngine::RedisPool;
using TraderEngine::RedisConnection;

namespace MessageFramework {

/**
 * @brief 消息消费者（基于Redis Stream Consumer Group）
 *
 * 职责：
 * 1. 从Redis Stream读取消息（XREADGROUP命令）
 * 2. 反序列化消息（Header + Body）
 * 3. 分发到MessageDispatcher处理
 * 4. 确认消息（XACK命令）
 * 5. 处理未ACK消息（XPENDING + XCLAIM）
 *
 * 特性：
 * - 独立线程运行
 * - 支持Consumer Group（负载均衡）
 * - 支持消息过滤（consumerId匹配）
 * - 支持失败重试
 * - 自动处理超时消息
 */
class MessageConsumer {
public:
    /**
     * @brief 配置选项
     */
    struct Config {
        std::string streamName;         ///< 监听的Stream名称
        std::string consumerGroup;      ///< Consumer Group名称
        std::string consumerId;         ///< 消费者ID（实例标识）
        int batchSize = 10;             ///< 每次读取的消息数量
        int blockMs = 1000;             ///< 阻塞等待时间（毫秒）
        int pendingCheckInterval = 60;  ///< 检查XPENDING的间隔（秒）
        int pendingTimeout = 300;       ///< 未ACK消息超时时间（秒）
        bool autoCreateGroup = true;    ///< 自动创建Consumer Group
        bool filterByConsumerId = true; ///< 过滤consumerId不匹配的消息
        std::string groupStartId = "$"; ///< Consumer Group起始ID（"0"=从头，"$"=从尾，默认"$"）
    };

    /**
     * @brief 消息回调函数（可选，代替MessageDispatcher）
     */
    using MessageCallback = std::function<int(const Message&)>;

    /**
     * @brief 构造函数
     *
     * @param redisPool Redis连接池
     * @param config 配置
     * @param dispatcher 消息分发器（可选）
     */
    MessageConsumer(RedisPool* redisPool,
                    const Config& config,
                    std::shared_ptr<MessageDispatcher> dispatcher = nullptr);

    ~MessageConsumer();

    /**
     * @brief 启动消费者
     */
    void start();

    /**
     * @brief 停止消费者
     */
    void stop();

    /**
     * @brief 是否正在运行
     */
    bool isRunning() const { return m_running; }

    /**
     * @brief 设置消息回调（代替Dispatcher）
     */
    void setMessageCallback(MessageCallback callback);

    /**
     * @brief 注册消息Handler（快捷方式）
     */
    void registerHandler(MessageHandlerPtr handler);

    /**
     * @brief 更新配置
     */
    void updateConfig(const Config& config);

    /**
     * @brief 获取统计信息
     */
    struct Stats {
        uint64_t messagesReceived = 0;
        uint64_t messagesProcessed = 0;
        uint64_t messagesFailed = 0;
        uint64_t messagesFiltered = 0;
        uint64_t pendingReclaimed = 0;
    };

    Stats getStats() const;

private:
    RedisPool* m_redisPool;
    Config m_config;
    std::shared_ptr<MessageDispatcher> m_dispatcher;
    MessageCallback m_messageCallback;

    std::atomic<bool> m_running;
    std::thread m_consumerThread;
    std::thread m_pendingThread;

    Stats m_stats;

    /**
     * @brief 消费者主循环
     */
    void consumerLoop();

    /**
     * @brief Pending消息处理循环
     */
    void pendingLoop();

    /**
     * @brief 创建Consumer Group
     */
    bool createConsumerGroup();

    /**
     * @brief 读取消息
     * @return vector of (Message, streamId) pairs
     */
    std::vector<std::pair<Message, std::string>> readMessages(std::shared_ptr<RedisConnection> conn);

    /**
     * @brief 处理单条消息
     */
    bool processMessage(std::shared_ptr<RedisConnection> conn, const Message& message, const std::string& streamId);

    /**
     * @brief 确认消息
     */
    bool ackMessage(std::shared_ptr<RedisConnection> conn, const std::string& streamId);

    /**
     * @brief 检查并处理Pending消息
     */
    void checkPendingMessages();

    /**
     * @brief 过滤消息（根据consumerId）
     */
    bool shouldProcessMessage(const MessageHeader& header) const;
};

using MessageConsumerPtr = std::shared_ptr<MessageConsumer>;

} // namespace MessageFramework
